package com.tianhao.rabbitmq.consumer;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeUnit;

/**
 * @ClassName: RabbitConsumer
 * @Description: 消费者
 * @author: tianhao.luo@hand-china.com
 * @date: 2020/11/10  22:08
 */
public class RabbitConsumer {
    private static final String QUEUE_NAME = "queue_demo";
    private static final String IP_ADDRESS = "192.168.15.78";
    private static final int PORT = 5672;

    public static void main(String[] args) throws Exception{
        /*
        网络地址的表示形式，即主机/端口对带有一些用于解析地址字符串的实用程序功能。
         */
        Address[] addresses = {new Address(IP_ADDRESS, PORT)};

        ConnectionFactory factory = new ConnectionFactory();

        factory.setUsername("root");
        factory.setPassword("root");
        /*
        创建一个新的代理连接，从列表中选择第一个可用地址。
         */
        Connection connection = factory.newConnection(addresses);
        final Channel channel = connection.createChannel();
        /*
        请求此通道的特定prefetchCount“服务质量”设置。
        设置客户端最多接收未被ack的消息的个数
         */
        channel.basicQos(64);
        /*
        创建一个默认的消费者，并重写接收到消息时的回调函数
         */
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("recv message:" + new String(body));

                try {
                    TimeUnit.SECONDS.sleep(1);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                /*
                 * 确认接收到的消息
                 * multiple:如果应该重新排队被拒绝的消息，而不是被丢弃/按照字母顺序排列，则为true
                 */
                channel.basicAck(envelope.getDeliveryTag(), false);
                System.out.println("----------------异步执行----------");
            }
        };
        /*
        使用显式确认和服务器生成的consumerTag启动非本地，非排他的使用者。
         */
        System.out.println("---------d----------------");
        /*
        该方法是异步执行的。
         */
        channel.basicConsume(QUEUE_NAME,defaultConsumer);
        System.out.println("---------e----------------");

        TimeUnit.SECONDS.sleep(5);

        channel.close();
        connection.close();
    }
}
